package Spark高级特性.累加器

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.util.AccumulatorV2
import org.junit.Test
import scala.collection.mutable


/**
 *  自定义累加器
 */
class Accumulator {


  /**
   * 将 RDD -> (1,2,3,4,5,6) 累加到一个集合中或是一个字符串
   */
  @Test
  def test(): Unit ={

    val conf = new SparkConf().setMaster("local[6]").setAppName("accumulator")
    val sc   = new SparkContext(conf)

    // 创建一个累加器对象
    val numAcc = new numAcc
    // 注册给 Spark
    sc.register(numAcc,"num")

    sc.parallelize(Seq("1","2","3","4","5")).foreach( item => numAcc.add(item))

    // 打印
    println(numAcc.value)

    // 关闭sc
    sc.stop()
  }
}
class numAcc extends AccumulatorV2[String ,Set[String]]{

  // 创建Set类型的nums作为结果返回
  private val nums:mutable.Set[String] = mutable.Set()  // mutable 将变量更改为可变

  // 初始的时候系统判断当前累加器是否为空
  override def isZero: Boolean = {
    nums.isEmpty
  }

  // 提供给Spark框架一个相同的拷贝累加器
  // 创建新的累加器
  override def copy(): AccumulatorV2[String, Set[String]] = {
    val newnumAcc = new numAcc()
    // 使用功能创建的对象作为同步锁保证线程安全
    nums.synchronized{
      newnumAcc.nums ++= this.nums
    }
    newnumAcc
  }

  // 提供给外部清理内容的方法
  // 帮助spark框架清理累加器的内容
  override def reset(): Unit = {
    nums.clear()
  }

  // 累加，外部传入要累加的内容在这个方法中累加
  // 一个集合和另一个集合相加 ---> ++=   一个集合和另一个值相加 ----> +=
  override def add(v: String): Unit = {
    nums += v
  }

  // 合并
  // 累加器在进行累加的时候可能每个分布式节点都会有一个实例
  // 在最后 Driver 进行一次合并，把所有实例的内容合并起来，会调用该方法
  override def merge(other: AccumulatorV2[String, Set[String]]): Unit = {
    nums ++= other.value
  }

  // 返回一个不可变的
  // 提供给外部一个累加结果
  // 因为外部有可能会再进行修改，如果是可变的很有可能会对内部的值产生影响
  override def value: Set[String] = {
    nums.toSet
  }

}